package com.tianlh.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.apache.curator.retry.ExponentialBackoffRetry;

/**
 * @author 黑子
 * @Date 2018/1/10 13:50
 */
public class ZkBarrier {
    /*
    * 可以看到多线程在并发情况下,都会准确地等待所有线程都处于就绪状态后才开始同时执行其他业务逻辑。如果是在同一个JVM中的话,
    * 使用CyclicBarrie完全可以解决诸如此类的多钱程同步问题。但是,如果是在分布式环境中可以使用Curator中提供的DistributedBarrier来解决。
    * */

    static String path = "/zk_barrier";
    static DistributedBarrier barrier;
    static DistributedDoubleBarrier doubleBarrier;

    public static void main(String[] args) throws Exception {

        /*
        * 1.手动解除 Barrier:
        * 我们模拟了5个线程,通过调用DistributedBarrier的setBarrier方法来完成Barrier的设置,
        * 并通过调用DistributedBarrier的waitOnBarrier方法来等待Barrier的释放。然后在主线程中,
        * 通过调用DistributedBarrier的removeBarrier方法来释放Barrier,同时触发所有等待该Barrier的5个线程同时进行各自的业务逻辑。
        *
        * */
        for (int i = 0; i < 5; ++i) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        RetryPolicy retry = new ExponentialBackoffRetry(1000, 3);
                        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
                            .sessionTimeoutMs(5000).retryPolicy(retry).build();
                        client.start();
                        barrier = new DistributedBarrier(client, path);
                        System.out.println(Thread.currentThread().getName() + "设置了Barrier");
                        barrier.setBarrier();
                        // 进入等待的状态
                        barrier.waitOnBarrier();

                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "thread-" + i).start();
        }
        //remove Barrier 所有处于等待的Barrier同时开始执行
        barrier.removeBarrier();

        /*
        * 2.自动解除 Barrier:
        * 指定了进入Barrier的成员数阀值,例如上面示例程序中的"5"。每个Barrier的参与者都会在调用方法之后进行等待,此时处于准备进入状态。
        * 一旦准备进入Barrier的成员数达到5个后，所有的成员会被同时触发进入。之后调用方法则会再次等待，此时处于准备退出状态。
        * 一旦准备退出Barrier的成员数达到5个后，所有的成员同样会被同时触发退出。
        *
        * */

        for (int i = 0; i < 5; ++i) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        RetryPolicy retry = new ExponentialBackoffRetry(1000, 3);
                        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
                            .sessionTimeoutMs(5000).retryPolicy(retry).build();
                        client.start();
                        doubleBarrier = new DistributedDoubleBarrier(client, path, 5);
                        Thread.sleep(Math.round(Math.random()*3000));
                        System.out.println(Thread.currentThread().getName() + "进程进入了Barrier");

                        // 等待进入状态
                        doubleBarrier.enter();
                        // 业务逻辑
                        System.out.println("启动...");
                        Thread.sleep(Math.round(Math.random()*3000));
                        // 等待离开状态
                        doubleBarrier.leave();
                        System.out.println("退出...");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "thread-" + i).start();
        }
    }
}
